import json
import os.path
from datetime import datetime
import pandas as pd
from kafka import KafkaProducer
import time
from config import ConfigData


def get_files_containing_substring(directory, substring):
    file_name = None
    # 遍历指定目录下的所有文件
    for filename in os.listdir(directory):
        # 构建完整的文件路径
        full_path = os.path.join(directory, filename)
        # 检查是否为文件
        if os.path.isfile(full_path):
            # 检查文件名是否包含指定的子字符串
            if substring in filename:
                file_name = full_path
    return file_name


def get_new_config(dir_path, count):
    json_file_path = os.path.join(dir_path, os.path.basename(dir_path)+".json")
    dir_name_list = os.path.basename(dir_path).split('-')
    file = open(json_file_path, 'r', encoding='utf-8')
    data = json.load(file)
    print(data)

    new_config = ConfigData.copy()
    new_config["gantry_info"] = {
        str(int(1000 + count)): {
          "up_gantry_id": dir_name_list[1],
          "down_gantry_id": dir_name_list[2],
          "direction": 0,
          "lane_num": 2,
          "time_interval": 1,
          "time_move": 0.25,
          "dist": data["kmNumber"],
          "car_pass_time": 2,
          "car_max_speed": 120,
          "car_min_speed": 70
        }
    }
    return new_config


def csv_to_list(path):
    # 示例数据
    df = pd.read_csv(path)
    df_data = df.to_dict(orient='records')
    # 创建一个空列表用于保存所有记录
    all_records = []
    # 定义门架ID
    gantry_a_id = os.path.basename(path).split('.')[0].split('-')[1]
    gantry_b_id = os.path.basename(path).split('.')[0].split('-')[2]

    # 遍历每一行数据
    for row in df_data:
        vlp = row['vlp']
        feevehicletype = row['feevehicletype']
        # 如果有经过门架A的时间，则添加一条记录
        if pd.notna(row['transtime_up']):
            transtime = pd.to_datetime(row['transtime_up'], format='mixed')
            record = {'gantry_id': gantry_a_id, 'vlp': vlp, 'feevehicletype': feevehicletype, 'transtime': transtime}
            all_records.append(record)

        # 如果有经过门架B的时间，则添加一条记录
        if pd.notna(row['transtime_down']):
            transtime = pd.to_datetime(row['transtime_down'], format='mixed')
            record = {'gantry_id': gantry_b_id, 'vlp': vlp, 'feevehicletype': feevehicletype, 'transtime': transtime}
            all_records.append(record)

    # 按照过车时间对记录进行排序
    all_records_sorted = sorted(all_records, key=lambda x: x['transtime'])
    return all_records_sorted


def send_csv_to_kafka(dir_path):
    current_time = None
    KAFKA_HOST = "106.120.201.126:19359"    # 公网kafka
    PRE_DATA_TOPIC = "PRE_DATA_TOPIC1111"
    # 发送数据
    producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                             value_serializer=lambda x: json.dumps(x).encode('utf-8'))
    csv_file_path = get_files_containing_substring(dir_path, 'all_')
    all_records_sorted = csv_to_list(csv_file_path)
    for row in all_records_sorted:
        msg = {
            "timestamp": row["transtime"].strftime("%Y-%m-%dT%H:%M:%S.%f"),
            "sn": row["gantry_id"],
            "plateNumber": row["vlp"],
            "vehicleType": row["feevehicletype"]
        }
        # 离线数据模拟发送在线数据
        producer.send(PRE_DATA_TOPIC, value=msg, key="key")
        print("发送数据", msg)
        t = datetime.strptime(msg["timestamp"], "%Y-%m-%dT%H:%M:%S.%f")
        if current_time is not None:
            sleep_duration = (t - current_time).total_seconds()
            if sleep_duration > 0:
                # print("当前时间:", t, " 跳过:", sleep_duration)
                # time.sleep(sleep_duration)
                time.sleep(0.2)
        current_time = t

send_csv_to_kafka("/home/gj/Download/新建文件夹")








